fetch_new_announcements.py

Fetches live corporate announcements for all stocks from the Dhan announcements API

Overview

Fetches live corporate announcements (regulatory filings, events) for all stocks in the master ISIN list using the Dhan Static ScanX API. Results are aggregated into a single JSON file sorted by date.

Source: fetch_new_announcements.py
Phase: Phase 2 (Enrichment)
Output: all_company_announcements.json

API Endpoint

POST https://ow-static-scanx.dhan.co/staticscanx/announcements

Request Payload

data.isin · string · required

The ISIN code of the security to fetch announcements for

Example Request

{"data": {"isin": "INE467B01029"}}

Function Signature

def fetch_announcements(item):
    """
    Fetch announcements for a single ISIN
    
    Args:
        item (dict): Stock item with Symbol, ISIN, and Name keys
    
    Returns:
        list: List of announcement dictionaries, or None if request fails
    """

Parameters

item.Symbol · string · required

Stock trading symbol (e.g., “TATAPOWER”)

item.ISIN · string · required

ISIN code used for API request

item.Name · string · required

Company display name

Output Structure

Symbol · string

Stock trading symbol

Name · string

Company name

Event · string

Announcement event description

Date · string

Announcement date

Type · string

Announcement type/category

Example Output

[
  {
    "Symbol": "TATAPOWER",
    "Name": "Tata Power Company Limited",
    "Event": "Board Meeting Intimation",
    "Date": "2024-01-15",
    "Type": "Regulatory"
  }
]

Dependencies

• master_isin_map.json — Input file containing all stock ISINs
• pipeline_utils.py — Provides BASE_DIR and get_headers() function
• requests — HTTP client library

Configuration

MAX_THREADS · int · default: "40"

Number of concurrent threads for parallel API requests

INPUT_FILE · string · default: "master_isin_map.json"

Path to master ISIN mapping file

OUTPUT_FILE · string · default: "all_company_announcements.json"

Output file path for aggregated announcements

Threading Implementation

Uses Python’s concurrent.futures.ThreadPoolExecutor with 40 workers for high-throughput parallel fetching:

with ThreadPoolExecutor(max_workers=MAX_THREADS) as executor:
    future_to_stock = {executor.submit(fetch_announcements, stock): stock for stock in master_list}
    
    completed = 0
    for future in as_completed(future_to_stock):
        res = future.result()
        if res:
            all_results.extend(res)
        completed += 1
        if completed % 100 == 0:
            print(f"Progress: {completed}/{len(master_list)} done.")

Error Handling

• Returns None if HTTP request fails or times out (10s timeout)
• Silently skips failed requests (no exception propagation)
• Validates response contains data field as a list
• Progress tracking: prints status every 100 completed requests

Post-Processing

Results are sorted by date in descending order before saving:

all_results.sort(key=lambda x: x.get("Date", ""), reverse=True)

Usage Example

python3 fetch_new_announcements.py

Expected Output:

Starting fetch for 2775 stocks...
Progress: 100/2775 done.
Progress: 200/2775 done.
...
Successfully saved 15432 announcements to all_company_announcements.json

Integration

This script is part of Phase 2 (Enrichment) in the EDL Pipeline. The output file is consumed by:

• add_corporate_events.py — Injects announcements into the “Recent Announcements” field of the final JSON

Run via master pipeline:

python3 run_full_pipeline.py